package com.fw.kafka_chapter_1.quick.interceptor;

import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @Author:yanwei
 * @Date: 2020/8/23 - 22:19
 */

public class ConcumerInterceptorShould implements ConsumerInterceptor<String,String> {



    /**
     * 空实现即可
     * @param map
     */
    @Override
    public void configure(Map<String, ?> map) {

    }

    // 拦截消息
    @Override
    public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> consumerRecords) {
        Map<TopicPartition, List<ConsumerRecord<String, String>>> newRecords = new HashMap<>();
        for (TopicPartition tp : consumerRecords.partitions()) {
            List<ConsumerRecord<String, String>> tpRecords = consumerRecords.records(tp);
            List<ConsumerRecord<String, String>> newTpRecords = new ArrayList<>();
            for (ConsumerRecord<String, String> record : tpRecords) {
                String interceptor = record.value().concat("interceptor");

                newTpRecords.add(new ConsumerRecord(record.topic(), record.partition(), record.offset(), record.key(), interceptor));
            }
            newRecords.put(tp, newTpRecords);
        }
        return new ConsumerRecords<>(newRecords);
    }

    // 消息提交的时候触发
    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> map) {

    }

    //  该消息生命流程结束后
    @Override
    public void close() {

    }
}
